package com.yifeng.repo.flink.data.transport.streaming.functions.task;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;

import com.google.common.base.Strings;
import com.yifeng.repo.flink.data.transport.config.DataSourceConfig;
import com.yifeng.repo.flink.data.transport.dto.DataPacket;
import com.yifeng.repo.flink.data.transport.dto.DataRow;

public class OracleFullSyncTask<T> extends CommonFullSyncTask {

	public static final String TABLE_SPLIT = "\\.";

	public OracleFullSyncTask(DataSourceConfig dataSourceConfig) {
		super(dataSourceConfig);
	}

	@Override
	protected PreparedStatement setFetchDirection(PreparedStatement ps,int fetchSize) {
		try {
			ps.setFetchSize(fetchSize);
			ps.setFetchDirection(ResultSet.FETCH_FORWARD);
		} catch (SQLException e) {
		}
		return ps;
	}

	@Override
	public DataRow assembleDataRow(DataRow.Type type, String tableName, HashMap dataMap) {
		//处理tableName,按.进行分隔
		String[] values = splitTableName(tableName);
		return new DataRow(type, dataMap,values[0],values[1],dataSourceConfig.getDatabase());
	}

	@Override
	public DataPacket assembleDataPacket(int queueNo, long idx, List rows, String tableName, boolean end) {
		String[] values = splitTableName(tableName);
		return DataPacket.ofFull(queueNo,idx, rows,values[1],dataSourceConfig.getDatabase(),values[0],dataSourceConfig.getHostname(),end);
	}
	
	private String[] splitTableName(String tableName) {
		if(Strings.isNullOrEmpty(tableName)) {
			throw new RuntimeException(" The tableName check error !" + tableName); 
		}
		String[] values = tableName.split(TABLE_SPLIT);
		if(values.length <= 1) {
			throw new RuntimeException(" The tableName check error !" + tableName);
		}
		return values;
	}
}
